package com.rxjava;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

//switchOnNext操作符是把一组Observable转换成一个Observable，转换规则为：对于这组Observable中的每一个Observable所产生的结果，如果在同一个时间内存在两个或多个Observable提交的结果，只取最后一个Observable提交的结果给订阅者
public class SwitchOnNextDemo {

	public static void main(String[] args) {
		//每隔500毫秒产生一个observable
        Observable<Observable<Long>> observable = Observable.interval(0, 500, TimeUnit.MILLISECONDS, Schedulers.trampoline()).map(new Function<Long, Observable<Long>>() {
            @Override
            public Observable<Long> apply(Long aLong) {
                //每隔200毫秒产生一组数据（0,10,20,30,40)
                return Observable.interval(0, 200, TimeUnit.MILLISECONDS).map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) {
                        return aLong * 10;
                    }
                }).take(5);
            }
        }).take(2);

        Observable.switchOnNext(observable).subscribe(new Observer<Long>() {
            @Override
            public void onComplete() {
                System.out.println("Sequence complete.");
            }

            @Override
            public void onError(Throwable e) {
                System.err.println("Error: " + e.getMessage());
            }

            @Override
            public void onNext(Long aLong) {
                System.out.println("Next:" + aLong);
            }

			@Override
			public void onSubscribe(Disposable disposable) {

			}
        });
	}

}
